<?php


namespace coco\queue;


use coco\Config;
use coco\Log;
use coco\client\Redis;

final class Job
{
    /**@var Redis $redis */
    private static $redis = null;

    private static function redisInstance()
    {
        if (empty(self::$redis)) {
            $queue_config = Config::getInstance()->get('queue');
            $reids_config = Config::getInstance()->get('redis');
            $host = $queue_config['redis_host'] ?? $reids_config['host'] ?? '127.0.0.1';
            $option = [
                'port' => $queue_config['redis_port'] ?? $reids_config['port'] ?? 6379,
                'password' => $queue_config['redis_password'] ?? $reids_config['password'] ?? '',
                'db' => $queue_config['redis_db'] ?? $reids_config['db'] ?? '',
                'reconnect' => $queue_config['redis_reconnect'] ?? $reids_config['reconnect'] ?? 10,
            ];
            self::$redis = new Redis($host, $option);
        }
    }

    /**
     * @param callable $caller callable用来提示编辑器
     * @param array $param
     * @param string $queue
     */

    public static function push(array $caller, array $param = [], string $queue = 'default')
    {
        $data = [
            'caller' => $caller,
            'param' => $param,
        ];
        self::redisInstance();
        self::$redis->lPush('coco_queue_' . $queue . ':running', serialize($data));
    }

    /**
     * @param int $do_at 执行时间戳
     * @param callable $caller callable用来提示编辑器
     * @param array $param
     * @param string $queue
     * @param bool $is_repeat false  是否可以重复插入 如果参数一样 默认不能重复
     * @return bool
     */
    public static function delay(int $do_at, array $caller, array $param = [], string $queue = 'default', bool $is_repeat = false): bool
    {
        $data = [
            'caller' => $caller,
            'param' => $param,
        ];
        if ($is_repeat) {
            //去重
            $data['_repeat'] = uniqid(posix_getpid());
        }
        self::redisInstance();
        return self::$redis->zAdd('coco_queue_' . $queue . ':delay', $do_at, serialize($data)) === 1;
    }

    public static function getToRun($queue = 'default')
    {
        self::redisInstance();
        $delays = self::$redis->zRangeByScore('coco_queue_' . $queue . ':delay', 0, time());
        foreach ($delays as $delay) {
            if (1 === self::$redis->zRem('coco_queue_' . $queue . ':delay', $delay)) {
                self::$redis->lPush('coco_queue_' . $queue . ':running', $delay);
            }
        }
    }

    public static function pop($queue = 'default')
    {
        self::redisInstance();
        $job_str = self::$redis->lPop('coco_queue_' . $queue . ':running');
        if (empty($job_str)) {
            return null;
        }
        $job = unserialize($job_str);
        if (empty($job)) {
            //忽略
            Log::getInstanceByKey('queue')->write('unserialize fail of ' . $job_str);
            return null;
        }
        return $job;
    }

    public static function rePush($queue, array $job)
    {
        self::redisInstance();
        self::$redis->lPush('coco_queue_' . $queue . ':running', serialize($job));
    }

    public static function complete($queue, array $job)
    {
        Log::getInstanceByKey('queue')->write('coco_queue_' . $queue . ':complete' . serialize($job)); 
    }

    public static function fail($queue, array $job)
    {
        self::redisInstance();
        self::$redis->zAdd('coco_queue_' . $queue . ':fail', time(), serialize($job)) === 1;
    }

    public static function exception($queue, array $job)
    {
        self::redisInstance();
        self::$redis->zAdd('coco_queue_' . $queue . ':exception', time(), serialize($job)) === 1;
    }

    public static function freeRedis()
    {
        self::$redis = null;
    }

    private $param;
    private $queue;
    private $try_index;

    public function __construct(string $queue, array $param ,int $try_index = 0)
    {
        $this->param = $param;
        $this->queue = $queue;
        $this->try_index = $try_index;
    }

    /**
     * @return int
     */
    public function getTryIndex(): int
    {
        return $this->try_index;
    }

    /**
     * @return array
     */
    public function getParam(): array
    {
        return $this->param;
    }

    /**
     * @return string
     */
    public function getQueue(): string
    {
        return $this->queue;
    }
}
